package edu.ucla.cs.rpc.multicast.sequencer.moving;

import java.io.IOException;
import java.net.SocketAddress;
import java.util.concurrent.locks.Condition;

import edu.ucla.cs.rpc.multicast.handlers.MessageHandler;
import edu.ucla.cs.rpc.multicast.network.MessageReceiver;
import edu.ucla.cs.rpc.multicast.network.MulticastMessageSender;
import edu.ucla.cs.rpc.multicast.network.message.Message;
import edu.ucla.cs.rpc.multicast.sequencer.Sequencer;
import edu.ucla.cs.rpc.multicast.util.token.DuplicateEntryException;
import edu.ucla.cs.rpc.multicast.util.token.NetworkTokenManager;
import edu.ucla.cs.rpc.multicast.util.token.Token;

/**
 * A Moving Sequencer is a member of a group of sequencers. These sequencers
 * pass a Token object amongst themselves to distribute the load of sending and
 * sequencing messages to Destinations.
 * 
 * @author Philip Russell
 * @see Total Order Broadcast and Multicast Algorithms: Taxonomy and Survery,
 *      Defago et al
 * @see Section 4.2, page 12
 */
public class MovingSequencer implements Sequencer {

	/*
	 * The communications agent that manages sending, receiving the token in the
	 * ring of MovingSequencers
	 */
	private NetworkTokenManager tokenManager = null;

	/* A communications agent for relaying the messages to the Destinations */
	private MulticastMessageSender sender = null;

	/* A communications agent for receiving messages from Senders */
	private MessageReceiver receiver = null;

	/* A condition variable that represents holding the token in the ring */
	private Condition haveToken;

	private MovingSequencer() {
	}

	/**
	 * Primary constructor.
	 * 
	 * @param multicastManager
	 *            The communications agent that will manage relaying messages
	 *            between a MovingSequencer and a Destination.
	 * @param tokenAddress
	 *            The address that the token communications agent (TokenManager)
	 *            is bound to.
	 * @throws IOException
	 *             Thrown if a lower-level network error, such as failing to
	 *             create/bind a socket, occurs.
	 */
	public MovingSequencer(SocketAddress multicastManager,
			SocketAddress tokenAddress) throws IOException {

		this.sender = new MulticastMessageSender(multicastManager);
		this.tokenManager = new NetworkTokenManager(tokenAddress);
		this.receiver = new MessageReceiver(new MovingSequencerMessageHandler());
	}

	/**
	 * Initializes the token sending/receiving functionality.
	 * 
	 */
	public void start() {
		/* Start listening for the token */
		Thread tokenListener = new Thread(this.tokenManager);
		tokenListener.start();
		/*
		 * Condition variable that the TokenManager will notify when it receives
		 * the token
		 */
		haveToken = this.tokenManager.getCondition();
	}

	/**
	 * Sets the next MovingSequencer in the ring.
	 * 
	 * @param nextSequencer
	 *            The next MovingSequencer that this MovingSequencer will pass
	 *            the token to.
	 */
	public void setNextSequencer(SocketAddress nextSequencer) {
		this.tokenManager.setNextAddress(nextSequencer);
	}

	/**
	 * 
	 * @return The address that the token communications agent (TokenManager) is
	 *         operating on.
	 */
	public SocketAddress getTokenAddress() {
		return this.tokenManager.getAddress();
	}

	/**
	 * Sets the token for the token communications agent (TokenManager). Needed
	 * to initialize the token ring.
	 * 
	 * @param token
	 *            A token.
	 */
	public void setToken(Token token) {
		this.tokenManager.setToken(token);
	}

	/**
	 * @return The address that this MovingSequencer listens for messages on.
	 */
	public SocketAddress getSocketAddress() {
		return this.receiver.getSocketAddress();
	}

	/**
	 * Explicit shutdown mechanism. This causes the main work loop of the
	 * process to fail, terminating the thread. Use this method instead of
	 * Thread.join() to terminate the process.
	 */
	public void shutdown() {
		this.tokenManager.shutdown();
		this.receiver.shutdown();
		this.sender.shutdown();
	}

	/**
	 * Primary functionality of a MovingSequencer. Receives messages from
	 * Senders, processes them, and relays them to Destinations.
	 * 
	 * @author Philip Russell Chase Covello
	 * 
	 */
	private class MovingSequencerMessageHandler implements MessageHandler {
		public void receive(Message message) {
			try {
				/*
				 * Wait until we are notified that the token has been receieved.
				 */
				tokenManager.getLock().lock();

				/* While we don't have the token.... */
				while (!tokenManager.hasToken()) {
					/* The lock is released here */
					haveToken.await();
				}
				/* The lock is re-aquired here */

				/* We have the token at this point */
				Token tok = tokenManager.getToken();

				try {
					/*
					 * Attempt to add the message to the SET of unique messages
					 * carried by the token.
					 */
					tok.addMessage(message);

					/*
					 * Order the message using the sequence number stored in the
					 * token.
					 */
					long seqNum = tok.getSeqNum();
					message.setSeqNum(seqNum);

					/* Send the message, update the token's sequence number */
					sender.send(message);
					seqNum++;
					tok.setSeqNum(seqNum);
				} catch (DuplicateEntryException e) {
					/*
					 * Adding the message to the set of unique messages failed
					 * because it was a duplicate entry.
					 */
				} catch (IOException e) {
					/* Some serious lower-level problem occurred, bail */
					e.printStackTrace();
				} finally {
					/*
					 * Notify the token manager that we can pass the token to
					 * the next MovingSequencer in the ring
					 */
					tokenManager.getCondition().signal();
				}
			} catch (InterruptedException e) {
			}
			/* Release the lock on the token. */
			tokenManager.getLock().unlock();
		}
	}
}
